package com.lintf.batch.config;

import cn.hutool.core.date.DateUtil;
import com.lintf.batch.api.ThirdPartyStudentApi;
import com.lintf.batch.domain.Student;
import com.lintf.batch.service.StudentService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.partition.support.Partitioner;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.dao.DeadlockLoserDataAccessException;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.backoff.FixedBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @ author lintf
 * @ system Mac
 * @ data 2025/8/15
 * @ package_name com.lintf.batch.config
 * @ project_name my-diy-project
 * @ desc
 */
@Slf4j
@Configuration
public class BatchStudentConfig {

    @Autowired
    private JobRepository jobRepository;
    @Autowired
    private PlatformTransactionManager transactionManager;
    @Autowired
    private StudentService studentService;
    @Autowired
    private JobExecutionListener jobExecutionListener;
    @Autowired
    private StepExecutionListener stepExecutionListener;

    private final int PAGE_SIZE = 1000;
    private final int TOTAL_PAGES = 100; // 200条 / 每页10条 = 20页

    /**
     * 创建分区器，将总页数分割为多个分区任务
     *
     * @return 分区映射，每个分区包含一个页码
     */
    @Bean
    public Partitioner partitioner() {
        return gridSize -> {
            Map<String, ExecutionContext> partitionMap = new HashMap<>();

            for (int i = 0; i < TOTAL_PAGES; i++) {
                ExecutionContext context = new ExecutionContext();
                context.putInt("page", i);
                partitionMap.put("partition" + i, context);
            }

            return partitionMap;
        };
    }

    /**
     * 创建主控步骤，用于管理分区任务的执行
     *
     * @return 主控步骤对象
     */
    @Bean
    public Step masterStep() {
        return new StepBuilder("masterStep", jobRepository)
                .partitioner("studentPartitioner", partitioner())
                .step(studentSyncStep())
                .taskExecutor(batchExecutor())
                .build();
    }

    /**
     * 创建学生数据读取器，按页读取学生数据
     *
     * @param page 当前页码
     * @return 学生数据读取器
     */
    @Bean
    @StepScope
    public ItemReader<Student> studentReader(
            @Value("#{stepExecutionContext['page']}") Integer page) {

        return new ItemReader<>() {
            private int currentPage = page;
            private List<Student> currentPageData;
            private int index;

            @Override
            public Student read() {
                try {
                    if (currentPageData == null) {
                        currentPageData = ThirdPartyStudentApi.fetchStudentsByBatch(currentPage, PAGE_SIZE);
                        if (currentPageData.isEmpty()) {
                            return null;
                        }
                        index = 0;
                    }

                    if (index < currentPageData.size()) {
                        return currentPageData.get(index++);
                    }

                    return null;
                } catch (Exception e) {
                    studentService.recordError("读取第" + currentPage + "页数据时出错", e);
                    return null;
                }
            }
        };
    }

    /**
     * 创建学生数据处理器，处理学生数据并设置最后同步时间
     *
     * @return 学生数据处理器
     */
    @Bean
    public ItemProcessor<Student, Student> studentItemProcessor() {
        return student -> {
            try {
                // 模拟处理逻辑，例如数据转换或验证
                student.setLastSyncTime(new Date());
                return student;
            } catch (Exception e) {
                studentService.recordError("处理学生数据时出错", e);
                throw e; // 让Spring Batch知道处理失败
            }
        };
    }

    /**
     * 创建学生数据写入器，将学生数据保存到数据库
     *
     * @return 学生数据写入器
     */
    @Bean
    public ItemWriter<Student> studentWriter() {
        return chunk -> {
            try {
                studentService.saveStudents((List<Student>) chunk.getItems());
            } catch (Exception e) {
                studentService.recordError("写入数据时出错", e);
                throw e;
            }
        };
    }

    /**
     * 创建学生数据处理器，处理学生数据并设置最后同步时间
     *
     * @return 学生数据处理器
     */
//    @Bean
//    public ItemProcessor<Student, Student> studentItemProcessor() {
//        return student -> {
//            // 检查是否已存在，避免重复处理
//            if (studentService.existsById(student.getStudentId())) {
//                return null;
//            }
//
//            // 使用重试模板处理可能失败的操作
//            return retryTemplate().execute(context -> {
//                try {
//                    // 模拟处理逻辑
//                    student.setLastSyncTime(new Date());
//
//                    // 模拟处理失败
//                    double random = Math.random();
//                    log.info("Processing student {} - random value: {}", student.getStudentId(), random);
//                    if (random < 0.05) { // 5%失败率
//                        throw new RuntimeException("数据处理失败");
//                    }
//
//                    return student;
//                } catch (Exception e) {
//                    log.warn("数据处理失败，尝试重试 ({}/3)", context.getRetryCount() + 1);
//                    throw e;
//                }
//            });
//        };
//    }
//
//    /**
//     * 创建学生数据写入器，将学生数据保存到数据库
//     *
//     * @return 学生数据写入器
//     */
//    @Bean
//    public ItemWriter<Student> studentWriter() {
//        return chunk -> {
//            try {
//                // 过滤已存在记录
//                List<Student> newStudents = chunk.getItems().stream()
//                        .filter(s -> !studentService.existsById(s.getStudentId()))
//                        .collect(Collectors.toList());
//
//                if (!newStudents.isEmpty()) {
//
//                    // 使用重试模板处理可能失败的操作
//                    retryTemplate().execute((RetryCallback<Void, RuntimeException>) context -> {
//                        try {
//                            studentService.saveStudents((List<Student>) chunk.getItems());
//
//                            // 模拟写入失败
//                            if (Math.random() < 0.02) { // 2%失败率
//                                throw new DeadlockLoserDataAccessException("数据库死锁", null);
//                            }
//
//                            return null;
//                        } catch (Exception e) {
//                            log.warn("数据写入失败，尝试重试 ({}/3)", context.getRetryCount() + 1);
//                            throw e;
//                        }
//                    });
//                }
//            } catch (Exception e) {
//                studentService.recordError("写入数据时出错", e);
//                throw e;
//            }
//        };
//    }


    /**
     * 创建自定义重试模板
     * <p>
     * return 重试模板对象
     */
    @Bean
    public RetryTemplate retryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();

        // 重试策略：最多重试3次
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(3);
        retryTemplate.setRetryPolicy(retryPolicy);

        // 退避策略：每次重试间隔500ms
        FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
        backOffPolicy.setBackOffPeriod(100);
        retryTemplate.setBackOffPolicy(backOffPolicy);

        return retryTemplate;
    }

    /**
     * 创建学生同步步骤，定义读取-处理-写入的批处理流程
     *
     * @return 学生同步步骤对象
     */
    @Bean
    public Step studentSyncStep() {
        return new StepBuilder("studentSyncStep", jobRepository)
                .<Student, Student>chunk(500, transactionManager)
                .reader(studentReader(null)) // 从分区获取页码
                .processor(studentItemProcessor())
                .writer(studentWriter())
                .faultTolerant()
                .retryLimit(2)
                .retry(Exception.class)
                .skip(Exception.class)
                .skipLimit(50)
                .listener(stepExecutionListener) // 添加步骤执行监听器
                .build();
    }

    /**
     * 创建学生数据同步作业，包含完整的批处理流程
     *
     * @return 学生数据同步作业对象
     */
    @Bean
    public Job syncStudentsJob() {
        return new JobBuilder("syncStudentsJob" + "_" + DateUtil.current(), jobRepository)
                .incrementer(new RunIdIncrementer())
                .listener(jobExecutionListener) // 添加作业执行监听器
                .start(masterStep())
                .build();
    }

    /**
     * 创建批处理线程池执行器
     *
     * @return 线程池执行器
     */
    @Bean
    public TaskExecutor batchExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(8); // 增加核心线程数
        executor.setMaxPoolSize(16); // 增加最大线程数
        executor.setQueueCapacity(100); // 增加队列容量
        return executor;
    }

}
